package com.cloud.stream.reciever;

import com.cloud.stream.model.Student;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

/**
 * @author shengtao
 * @Description:
 * @date 2019/06/04 11:32
 */
@EnableBinding(Sink.class)
public class Reciever {

    @StreamListener(Sink.INPUT)
    public void recieve(Message<Student> message) {
        Student student = message.getPayload();
        System.out.println("===============:" + student);
        MessageHeaders messageHeaders = message.getHeaders();
        Object me =  messageHeaders.get(KafkaHeaders.TIMESTAMP);
        if(me != null){

        }
//        Acknowledgment acknowledgment = messageHeaders.get(KafkaHeaders.ACKNOWLEDGMENT,Acknowledgment.class);
//        if (acknowledgment != null) {
//            System.out.println("Acknowledgment provided");
//            acknowledgment.acknowledge();
//        }
//        String id = messageHeaders.get(KafkaHeaders.CORRELATION_ID).toString();
//        messageHeaders.get(KafkaHeaders.RAW_DATA).toString();
//        System.out.println("=============== id:" + messageHeaders.getId() + " ========= timstamp:"+messageHeaders.getTimestamp() + " ============ key:"+messageHeaders.get("key"));

    }

}
